package com.example.day03.listener;

//import com.ververica.cdc.connectors.mysql.source.MySqlSource;
//import com.ververica.cdc.connectors.mysql.table.StartupOptions;
//import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import com.example.day03.sink.MysqlDataChangeSink;
import org.apache.flink.cdc.connectors.mysql.source.MySqlSource;
import org.apache.flink.cdc.connectors.mysql.table.StartupOptions;
import org.apache.flink.cdc.debezium.JsonDebeziumDeserializationSchema;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;

import java.util.Properties;

/**
 * 参考
 * @link <a href="https://nightlies.apache.org/flink/flink-cdc-docs-master/zh/docs/connectors/flink-sources/mysql-cdc/">...</a>
 */
@Component
public class MysqlEventListener implements ApplicationRunner {

    @Autowired
    private MysqlDataChangeSink dataChangeSink;

    public MysqlEventListener(MysqlDataChangeSink dataChangeSink) {
        this.dataChangeSink = dataChangeSink;
    }

    @Override
    public void run(ApplicationArguments args) throws Exception {
        //Apache Flink Dashboard 配置
        Configuration config = new Configuration();
        config.set(RestOptions.PORT, 9090);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(config);
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        MySqlSource<String> dataChangeInfoMySqlSource = buildDataChangeSource();
        env.fromSource(dataChangeInfoMySqlSource, WatermarkStrategy.noWatermarks(), "mysql-source")
                // 设置 source 节点的并行度为 4
                .setParallelism(4)
                //添加输出任务
                .addSink(dataChangeSink)
                // 设置 sink 节点并行度为 1
                .setParallelism(1);
        env.execute("mysql-stream-cdc");
    }

    private MySqlSource<String> buildDataChangeSource() {
        Properties prop = new Properties();
        prop.put("decimal.handling.mode", "string");
        return MySqlSource.<String>builder()
                .hostname("node3")
                .port(3306)
                .scanNewlyAddedTableEnabled(true)// 启用扫描新添加的表功能
                .username("root")
                .debeziumProperties(prop)
                .password("123456")
                .databaseList("mydb")//设置捕获的数据库， 如果需要同步整个数据库，请将 tableList 设置为 ".*".
                .tableList("mydb.customers")// 设置捕获的表
                /*
                 * initial初始化快照,即全量导入后增量导入(检测更新数据写入)
                 * latest:只进行增量导入(不读取历史变化)
                 * timestamp:指定时间戳进行数据导入(大于等于指定时间错读取数据)
                 */
                .startupOptions(StartupOptions.latest())
                .deserializer(new JsonDebeziumDeserializationSchema())// 将 SourceRecord 转换为 JSON 字符串
                .serverTimeZone("GMT+8")//指定mysql的时区
                .build();
    }
}
